package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/3/13 08:45
 */
public class Dwd_02_DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeCartAdd().init(3002, 2, "Dwd_02_DwdTradeCartAdd");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_02_DwdTradeCartAdd");
        
        // 2. 过滤 cart_info 数据
        Table cartAdd = tEnv.sqlQuery("select " +
                                          "data['id'] id," +
                                          "data['user_id'] user_id," +
                                          "data['sku_id'] sku_id," +
                                          "data['source_id'] source_id," +
                                          "data['source_type'] source_type," +
                                          "if(`type` = 'insert'," +
                                          "     data['sku_num']," +
                                          "     cast((cast(data['sku_num'] as int) - cast(`old`['sku_num'] as int)) as string)" +
                                          ") sku_num," +
                                          "ts," +
                                          "pt " +
                                          "from ods_db " +
                                          "where `database`='gmall2022' " +
                                          "and `table`='cart_info' " +
                                          "and (" +
                                          " `type`='insert' " +
                                          " or (`type`='update'" +
                                          "     and `old`['sku_num'] is not null " + // 表示 sku_num 发生了变化
                                          "     and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)" +  // 变大
                                          " )" +
                                          ")");
        tEnv.createTemporaryView("cart_add", cartAdd);
        
        //        cartAdd.execute().print();  // print 阻塞式: 会阻塞后面所有代码
        // 3. 读取 base_dic
        readBaseDic(tEnv);
        
        // 3. 事实表与维度表进行 lookup join
        Table result = tEnv.sqlQuery("select " +
                                        "ca.id," +
                                        "user_id," +
                                        "sku_id," +
                                        "source_id," +
                                        "source_type," +
                                        "dic_name source_type_name," +
                                        "sku_num," +
                                        "ts " +
                                        "from cart_add ca " +
                                        "join base_dic for system_time as of ca.pt as dic " +
                                        "on ca.source_type=dic.dic_code ");
    
        //result.execute().print();
        
        
        // 4. 把结果写出到kafka
        // 优先选择普通的 kafka, 只有当查询的结果中有 left(right,full) join, 和聚和的时候使用 upsert_kafka
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            "id string," +
                            "user_id string," +
                            "sku_id string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
    
        result.executeInsert("dwd_trade_cart_add");
    
    }
}
/*
交易域加购事务事实表
    数据源:
        ods 层 ods_db
        
        cart_info 表: insert update(sku_num 变大)
        
     user_id   sku_id   sku_num
       1        1        2        insert  直接写出
       1        1        3        update  直接  3-2=1(变量)
       
    
    表的 join:
        维度退化:
            事实表与维度表的 join
            
        
        事实表与事实表的 join
        

 */